/*
 * (C) Copyright 2014-2016, by Dimitrios Michail
 *
 * JHeaps Library
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.jheaps.tree

import org.jheaps.AddressableHeap
import org.jheaps.MergeableAddressableHeap
import org.jheaps.annotations.ConstantTime
import org.jheaps.annotations.LogarithmicTime
import java.io.Serializable

/**
 * Pairing heaps. The heap is sorted according to the [ natural ordering][Comparable] of its keys, or by a [Comparator] provided at heap
 * creation time, depending on which constructor is used.
 *
 *
 *
 * This implementation provides amortized O(log(n)) time cost for the
 * `insert`, `deleteMin`, and `decreaseKey` operations.
 * Operation `findMin`, is a worst-case O(1) operation. The algorithms are
 * based on the [pairing heap
 * paper](http://dx.doi.org/10.1007/BF01840439). Pairing heaps are very efficient in practice, especially in
 * applications requiring the `decreaseKey` operation. The operation
 * `meld` is amortized O(log(n)).
 *
 *
 *
 * All the above bounds, however, assume that the user does not perform
 * cascading melds on heaps such as:
 *
 * <pre>
 * d.meld(e);
 * c.meld(d);
 * b.meld(c);
 * a.meld(b);
</pre> *
 *
 * The above scenario, although efficiently supported by using union-find with
 * path compression, invalidates the claimed bounds.
 *
 *
 *
 * Note that the ordering maintained by a pairing heap, like any heap, and
 * whether or not an explicit comparator is provided, must be *consistent
 * with `equals`* if this heap is to correctly implement the
 * `AddressableHeap` interface. (See `Comparable` or
 * `Comparator` for a precise definition of *consistent with
 * equals*.) This is so because the `AddressableHeap` interface is
 * defined in terms of the `equals` operation, but a pairing heap performs
 * all key comparisons using its `compareTo` (or `compare`) method,
 * so two keys that are deemed equal by this method are, from the standpoint of
 * this heap, equal. The behavior of a heap *is* well-defined even if its
 * ordering is inconsistent with `equals`; it just fails to obey the
 * general contract of the `AddressableHeap` interface.
 *
 *
 *
 * **Note that this implementation is not synchronized.** If
 * multiple threads access a heap concurrently, and at least one of the threads
 * modifies the heap structurally, it *must* be synchronized externally.
 * (A structural modification is any operation that adds or deletes one or more
 * elements or changing the key of some element.) This is typically accomplished
 * by synchronizing on some object that naturally encapsulates the heap.
 *
 * @param <K>
 * the type of keys maintained by this heap
 * @param <V>
 * the type of values maintained by this heap
 *
 * @author Dimitrios Michail
 *
 * @see RankPairingHeap
 *
 * @see CostlessMeldPairingHeap
 *
 * @see FibonacciHeap
</V></K> */
class PairingHeap<K, V> @ConstantTime constructor(
    /**
     * The comparator used to maintain order in this heap, or null if it uses
     * the natural ordering of its keys.
     *
     * @serial
     */
    private val comparator: Comparator<in K?>?
) : MergeableAddressableHeap<K, V>,
    Serializable {

    /**
     * The root of the pairing heap
     */
    private var root: Node? = null

    /**
     * Size of the pairing heap
     */
    private var size: Long

    /**
     * Used to reference the current heap or some other pairing heap in case of
     * melding, so that handles remain valid even after a meld, without having
     * to iterate over them.
     *
     * In order to avoid maintaining a full-fledged union-find data structure,
     * we disallow a heap to be used in melding more than once. We use however,
     * path-compression in case of cascading melds, that it, a handle moves from
     * one heap to another and then another.
     */
    private var other: PairingHeap<K, V>

    /**
     * Constructs a new, empty heap, using the natural ordering of its keys. All
     * keys inserted into the heap must implement the [Comparable]
     * interface. Furthermore, all such keys must be *mutually
     * comparable*: `k1.compareTo(k2)` must not throw a
     * `ClassCastException` for any keys `k1` and `k2` in the
     * heap. If the user attempts to put a key into the heap that violates this
     * constraint (for example, the user attempts to put a string key into a
     * heap whose keys are integers), the `insert(Object key)` call will
     * throw a `ClassCastException`.
     */
    @ConstantTime
    constructor() : this(null)

    /**
     * Constructs a new, empty heap, ordered according to the given comparator.
     * All keys inserted into the heap must be *mutually comparable* by
     * the given comparator: `comparator.compare(k1,
     * k2)` must not throw a `ClassCastException` for any keys `k1`
     * and `k2` in the heap. If the user attempts to put a key into the
     * heap that violates this constraint, the `insert(Object key)` call
     * will throw a `ClassCastException`.
     *
     * @param comparator
     * the comparator that will be used to order this heap. If
     * `null`, the [natural ordering][Comparable] of
     * the keys will be used.
     */
    init {
        this.size = 0
        this.other = this
    }

    /**
     * {@inheritDoc}
     *
     * @throws IllegalStateException
     * if the heap has already been used in the right hand side of a
     * meld
     */
    @LogarithmicTime(amortized = true)
    override fun insert(key: K, value: V?): Node {
        check(other === this) { "A heap cannot be used after a meld" }
        if (key == null) {
            throw NullPointerException("Null keys not permitted")
        }
        val n = Node(this, key, value)
        root = if (comparator == null) {
            link(root, n)
        } else {
            linkWithComparator(root, n)
        }
        size++
        return n
    }

    /**
     * {@inheritDoc}
     *
     * @throws IllegalStateException
     * if the heap has already been used in the right hand side of a
     * meld
     */
    @LogarithmicTime(amortized = true)
    override fun insert(key: K): Node {
        return insert(key, null)
    }

    /**
     * {@inheritDoc}
     */
    @ConstantTime(amortized = false)
    override fun findMin(): Node {
        if (size == 0L) {
            throw NoSuchElementException()
        }
        return root!!
    }

    /**
     * {@inheritDoc}
     */
    @LogarithmicTime(amortized = true)
    override fun deleteMin(): Node{
        if (size == 0L) {
            throw NoSuchElementException()
        }

        // assert root.o_s == null && root.y_s == null;
        val oldRoot = root

        // cut all children, combine them and overwrite old root
        root = combine(cutChildren(root!!))

        // decrease size
        size--

        return oldRoot!!
    }

    @get:ConstantTime
    override val isEmpty: Boolean
        /**
         * {@inheritDoc}
         */
        get() = size == 0L

    /**
     * {@inheritDoc}
     */
    @ConstantTime
    override fun size(): Long {
        return size
    }

    /**
     * {@inheritDoc}
     */
    override fun comparator(): Comparator<in K?>? {
        return comparator
    }

    /**
     * {@inheritDoc}
     */
    @ConstantTime(amortized = false)
    override fun clear() {
        root = null
        size = 0
    }

    /**
     * {@inheritDoc}
     */
    @LogarithmicTime(amortized = true)
    override fun meld(other: MergeableAddressableHeap<K, V>) {
        val h = other as PairingHeap<K, V>

        // check same comparator
        if (comparator != null) {
            require(!(h.comparator == null || h.comparator != comparator)) { "Cannot meld heaps using different comparators!" }
        } else require(h.comparator == null) { "Cannot meld heaps using different comparators!" }

        check(h.other === h) { "A heap cannot be used after a meld." }

        // perform the meld
        size += h.size
        root = if (comparator == null) {
            link(root, h.root)
        } else {
            linkWithComparator(root, h.root)
        }

        // clear other
        h.size = 0
        h.root = null

        // take ownership
        h.other = this
    }

    // --------------------------------------------------------------------
    inner class Node(/*
         * We maintain explicitly the belonging heap, instead of using an inner
         * class due to possible cascading melding.
         */var heap: PairingHeap<K, V>,
           override var key: K,
           override var value: V?
    ) : AddressableHeap.Handle<K, V>, Serializable {
        /**
         * {@inheritDoc}
         */
        var o_c: Node? = null // older child
        var y_s: Node? = null // younger sibling
        var o_s: Node? = null // older sibling or parent

        /**
         * {@inheritDoc}
         */
        @LogarithmicTime(amortized = true)
        override fun decreaseKey(newKey: K) {
            owner.decreaseKey(this, newKey)
        }

        /**
         * {@inheritDoc}
         */
        @LogarithmicTime(amortized = true)
        override fun delete() {
            owner.delete(this)
        }

        /*
        * Get the owner heap of the handle. This is union-find with
        * path-compression between heaps.
        */
        val owner: PairingHeap<K, V>
            get() {
                if (heap.other !== heap) {
                    // find root
                    var root = heap
                    while (root !== root.other) {
                        root = root.other
                    }
                    // path-compression
                    var cur = heap
                    while (cur.other !== root) {
                        val next = cur.other
                        cur.other = root
                        cur = next
                    }
                    heap = root
                }
                return heap
            }
    }

    /**
     * Decrease the key of a node.
     *
     * @param n
     * the node
     * @param newKey
     * the new key
     */
    private fun decreaseKey(n: Node, newKey: K) {
        val c = comparator?.compare(newKey, n.key) ?: (newKey as Comparable<K?>).compareTo(n.key)

        require(c <= 0) { "Keys can only be decreased!" }
        n.key = newKey
        if (c == 0 || root === n) {
            return
        }

        requireNotNull(n.o_s) { "Invalid handle!" }

        // unlink from parent
        if (n.y_s != null) {
            n.y_s!!.o_s = n.o_s
        }
        if (n.o_s!!.o_c === n) { // I am the oldest :(
            n.o_s!!.o_c = n.y_s
        } else { // I have an older sibling!
            n.o_s!!.y_s = n.y_s
        }
        n.y_s = null
        n.o_s = null

        // merge with root
        root = if (comparator == null) {
            link(root!!, n)
        } else {
            linkWithComparator(root!!, n)
        }
    }

    /*
     * Delete a node
     */
    private fun delete(n: Node) {
        if (root === n) {
            deleteMin()
            n.o_c = null
            n.y_s = null
            n.o_s = null
            return
        }

        requireNotNull(n.o_s) { "Invalid handle!" }

        // unlink from parent
        if (n.y_s != null) {
            n.y_s!!.o_s = n.o_s
        }
        if (n.o_s!!.o_c === n) { // I am the oldest :(
            n.o_s!!.o_c = n.y_s
        } else { // I have an older sibling!
            n.o_s!!.y_s = n.y_s
        }
        n.y_s = null
        n.o_s = null

        // perform delete-min at tree rooted at this
        val t = combine(cutChildren(n))

        // and merge with other cut tree
        root = if (comparator == null) {
            link(root!!, t)
        } else {
            linkWithComparator(root!!, t)
        }

        size--
    }

    /*
     * Two pass pair and compute root.
     */
    private fun combine(l: Node?): Node? {
        if (l == null) {
            return null
        }

        assert(l.o_s == null)
        // left-right pass
        var pairs: Node? = null
        var it = l
        var p_it: Node?
        if (comparator == null) { // no comparator
            while (it != null) {
                p_it = it
                it = it.y_s

                if (it == null) {
                    // append last node to pair list
                    p_it.y_s = pairs
                    p_it.o_s = null
                    pairs = p_it
                } else {
                    val n_it = it.y_s

                    // disconnect both
                    p_it.y_s = null
                    p_it.o_s = null
                    it.y_s = null
                    it.o_s = null

                    // link trees
                    p_it = link(p_it, it)!!

                    // append to pair list
                    p_it.y_s = pairs
                    pairs = p_it

                    // advance
                    it = n_it
                }
            }
        } else {
            while (it != null) {
                p_it = it
                it = it.y_s

                if (it == null) {
                    // append last node to pair list
                    p_it.y_s = pairs
                    p_it.o_s = null
                    pairs = p_it
                } else {
                    val n_it = it.y_s

                    // disconnect both
                    p_it.y_s = null
                    p_it.o_s = null
                    it.y_s = null
                    it.o_s = null

                    // link trees
                    p_it = linkWithComparator(p_it, it)!!

                    // append to pair list
                    p_it.y_s = pairs
                    pairs = p_it

                    // advance
                    it = n_it
                }
            }
        }

        // second pass (reverse order - due to add first)
        it = pairs
        var f: Node? = null
        if (comparator == null) {
            while (it != null) {
                val nextIt = it.y_s
                it.y_s = null
                f = link(f, it)
                it = nextIt
            }
        } else {
            while (it != null) {
                val nextIt = it.y_s
                it.y_s = null
                f = linkWithComparator(f, it)
                it = nextIt
            }
        }

        return f
    }

    /**
     * Cut the children of a node and return the list.
     *
     * @param n
     * the node
     * @return the first node in the children list
     */
    private fun cutChildren(n: Node): Node? {
        val child = n.o_c
        n.o_c = null
        if (child != null) {
            child.o_s = null
        }
        return child
    }

    private fun link(f: Node?, s: Node?): Node? {
        if (s == null) {
            return f
        } else if (f == null) {
            return s
        } else if ((f.key as Comparable<K?>?)!!.compareTo(s.key) <= 0) {
            s.y_s = f.o_c
            s.o_s = f
            if (f.o_c != null) {
                f.o_c!!.o_s = s
            }
            f.o_c = s
            return f
        } else {
            return link(s, f)
        }
    }

    private fun linkWithComparator(f: Node?, s: Node?): Node? {
        if (s == null) {
            return f
        } else if (f == null) {
            return s
        } else if (comparator!!.compare(f.key, s.key) <= 0) {
            s.y_s = f.o_c
            s.o_s = f
            if (f.o_c != null) {
                f.o_c!!.o_s = s
            }
            f.o_c = s
            return f
        } else {
            return linkWithComparator(s, f)
        }
    }
}
